package com.example.rocket.consumer;

import com.aliyun.openservices.ons.api.*;
import com.example.rocket.core.NbMsg;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.core.env.Environment;
import org.springframework.util.Assert;

import java.util.Objects;
import java.util.Properties;

/**
 * @description: Ons-Consumer
 * @author TAO
 * @date 2021/12/22 4:04 下午
 */

@Slf4j
@Data
public class OnsNbMQConsumer extends AbstractNbMQConsumer {

    /**
     * ons 消费者
     */
    private Consumer consumer;

    /**
     * 初始化 MQ 消息监听器
     * @throws MQClientException
     */
    protected void initNbMQPushConsumer() throws MQClientException {
        if (nbMqListener == null) {
            throw new IllegalArgumentException("MQ消息消息监听器初始化需要rocketMQListener");
        }
        log.info("===>{}","initNbMQPushConsumer");
        Assert.notNull(consumerGroup, "属性“consumerGroup”是必需的");
        Assert.notNull(nameServer, "属性“nameServer”是必需的");
        Assert.notNull(topic, "属性“topic”是必需的");

        Properties properties = new Properties();


        Environment environment = applicationContext.getEnvironment();
        properties.put(PropertyKeyConst.GROUP_ID, consumerGroup);
        properties.put(PropertyKeyConst.AccessKey, environment.resolveRequiredPlaceholders(accessKey));
        properties.put(PropertyKeyConst.SecretKey, environment.resolveRequiredPlaceholders(secretKey));
        properties.put(PropertyKeyConst.NAMESRV_ADDR, nameServer);
        properties.put(PropertyKeyConst.MaxReconsumeTimes, 3);
        consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(topic,selectorExpression, new MessageListener() {
            @Override
            public Action consume(Message message, ConsumeContext consumeContext) {
                try {
                    handleMessage(new NbMsg(message));
                    //if(onMessage(message.getTopic(), message.getTag(),message.getMsgID(), getMessageObject(message.getBody()))){
                    //    return Action.CommitMessage;
                    //}else{
                    //    return Action.ReconsumeLater;
                    //}
                    return Action.CommitMessage;
                } catch (Exception e) {
                    log.error("消息id[{}],消费消息异常 : {}",message.getMsgID(),e);
                    return Action.ReconsumeLater;
                }
            }
        });
    }



    @Override
    public void destroy() throws Exception {
        this.setRunning(false);
        if (Objects.nonNull(consumer)) {
            consumer.shutdown();
        }
        log.info("ONS监听器容器 销毁, {}", this.toString());
    }

    @Override
    public void start() {
        log.info("ONS===>start");
        if (this.isRunning()) {
            throw new IllegalStateException("容器已经运行！！！ " + this.toString());
        }
        try {
            consumer.start();
        } catch (Exception e) {
            throw new IllegalStateException("OnsMQ失败！！！", e);
        }
        this.setRunning(true);

        log.info("OnsMQ运行成功！！！ {}", this.toString());
    }

    @Override
    public void stop() {
        log.info("ONS===>stop");
        if (this.isRunning()) {
            if (Objects.nonNull(consumer)) {
                consumer.shutdown();
            }
            setRunning(false);
        }
    }
}
